package com.ruoyi.flink.status;/**
 * @program: ruoyi
 * @description: FlinkStatusStudy
 * @author: zengchen
 * @create: 2021-02-28 10:20
 **/

import com.ruoyi.flink.bean.TaxiFare;
import com.ruoyi.flink.source.MyTaxiFareSource;
import lombok.AllArgsConstructor;
import lombok.Data;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.Optional;

/**
 * @program: ruoyi
 * @description: flink 状态管理学习
 * @author: zengchen
 * @create: 2021-02-28 10:20
 **/
public class FlinkStatusStudy {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
//        env.setRuntimeMode(RuntimeExecutionMode.BATCH);
        DataStreamSource<TaxiFare> taxiFareDataStreamSource = env.addSource(new MyTaxiFareSource());

        // 运费总额到达 500 的司机
        taxiFareDataStreamSource
                .map(new MapFunction<TaxiFare, Tuple2<Long, Float>>() {
                    @Override
                    public Tuple2<Long, Float> map(TaxiFare value) throws Exception {
                        return new Tuple2<Long, Float>(value.driverId, value.totalFare);
                    }
                })
//                .assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(20))
//                        .withTimestampAssigner((event, timestamp) -> event.f0))
                .keyBy(new KeySelector<Tuple2<Long, Float>, Long>() {
                    @Override
                    public Long getKey(Tuple2<Long, Float> value) throws Exception {
                        return value.f0;
                    }
                })
                .process(new KeyedProcessFunction<Long, Tuple2<Long, Float>, Tuple2<Long, Float>>() {
                    private transient ValueState<Float> total;
                    private transient ValueState<List<SomeBean>> beanTest;

                    @Override
                    public void processElement(Tuple2<Long, Float> element, Context ctx, Collector<Tuple2<Long, Float>> out) throws Exception {
                        Float newValue = Optional.ofNullable(total.value()).orElse(0F) + element.f1;
                        List<SomeBean> beans = beanTest.value();

                        if (CollectionUtils.isEmpty(beans)) {
                            beans = new ArrayList<>();
                        }
                        beans.add(new SomeBean(element.f0, element.f1));
                        if (newValue > 500) {
                            element.f1 = newValue;
                            out.collect(element);
                        }
                        total.update(newValue);
                        beanTest.update(beans);
                    }

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<Float> flagDescriptor = new ValueStateDescriptor<>(
                                "total",
                                Types.FLOAT);
                        ValueStateDescriptor<List<SomeBean>> SomeBeanDescriptor = new ValueStateDescriptor(
                                "someBean",
                                List.class);
                        total = getRuntimeContext().getState(flagDescriptor);
                        beanTest = getRuntimeContext().getState(SomeBeanDescriptor);
                    }
                })
                .print();

        env.execute("FlinkStatusStudy");

    }

    @Data
    @AllArgsConstructor
    static class SomeBean {
        private Long diverId;
        private Float all;
    }
}
